package org.gbif.metrics.cli;

import org.gbif.api.model.occurrence.Occurrence;
import org.gbif.common.messaging.api.MessageCallback;
import org.gbif.common.messaging.api.OccurrencePersistedMessage;
import org.gbif.metrics.cube.HBaseCubes;
import org.gbif.metrics.cube.occurrence.OccurrenceAddressUtil;
import org.gbif.metrics.cube.occurrence.OccurrenceCube;

import java.io.IOException;

import com.google.common.base.Throwables;
import com.urbanairship.datacube.Batch;
import com.urbanairship.datacube.ops.LongOp;
import org.apache.hadoop.conf.Configuration;

/**
 * The service to update the occurrence cube.
 */
class OccurrenceCubeUpdaterService extends CubeUpdaterService<CubeConfiguration> {

  public OccurrenceCubeUpdaterService(CubeConfiguration configuration) {
    super(configuration);
  }

  @Override
  MessageCallback<OccurrencePersistedMessage> getCallback(final CubeConfiguration configuration,
    final Configuration hadoopConfiguration) {

    return new CubeMutatorCallback<LongOp>() {

      // reduce reuse recycle HBase connections
      private volatile HBaseCubes<LongOp> singletonCube;

      @Override
      protected synchronized HBaseCubes<LongOp> getCube() {
        if (singletonCube == null) {
          try {
            singletonCube =
              HBaseCubes.newIncrementingBatchAsync(OccurrenceCube.INSTANCE, LongOp.DESERIALIZER,
                configuration.cubeTable.getBytes(), configuration.columnFamily.getBytes(), hadoopConfiguration,
                configuration.writeBatchSize);
          } catch (IOException e) {
            throw Throwables.propagate(e);
          }
        }
        return singletonCube;
      }

      @Override
      protected Batch<LongOp> getNewMutations(Occurrence occurrence) {
        return OccurrenceAddressUtil.cubeMutation(occurrence, new LongOp(1));
      }

      @Override
      protected Batch<LongOp> getUpdateMutations(Occurrence oldOccurrence, Occurrence newOccurrence) {
        Batch<LongOp> update = OccurrenceAddressUtil.cubeMutation(newOccurrence, new LongOp(1));
        update.putAll(OccurrenceAddressUtil.cubeMutation(oldOccurrence, new LongOp(-1)));
        return update;
      }
    };
  }
}
